package indi.zqc.mqtt.customer.configuration;

import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;

/**
 * TODO
 *
 * @author AZHE_SUN
 * @date 2021-4-2 10:46
 **/
public class ListConverter implements MqttMessageConverter {
    private final String charset;
    private final Integer defaultQos;
    private final Boolean defaultRetained;
    private volatile boolean payloadAsBytes;
    private volatile BeanFactory beanFactory;
    private volatile MessageBuilderFactory messageBuilderFactory;
    private volatile boolean messageBuilderFactorySet;

    public ListConverter() {
        this(0, false);
    }

    public ListConverter(int defaultQos, boolean defaultRetain) {
        this(defaultQos, defaultRetain, "UTF-8");
    }

    public ListConverter(String charset) {
        this(0, false, charset);
    }

    public ListConverter(int defaultQos, boolean defaultRetained, String charset) {
        this.payloadAsBytes = false;
        this.messageBuilderFactory = new DefaultMessageBuilderFactory();
        this.defaultQos = defaultQos;
        this.defaultRetained = defaultRetained;
        this.charset = charset;
    }

    public final void setBeanFactory(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
    }

    protected BeanFactory getBeanFactory() {
        return this.beanFactory;
    }

    protected MessageBuilderFactory getMessageBuilderFactory() {
        if (!this.messageBuilderFactorySet) {
            if (this.beanFactory != null) {
                this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
            }

            this.messageBuilderFactorySet = true;
        }

        return this.messageBuilderFactory;
    }

    public void setPayloadAsBytes(boolean payloadAsBytes) {
        this.payloadAsBytes = payloadAsBytes;
    }

    public boolean isPayloadAsBytes() {
        return this.payloadAsBytes;
    }

    public Message<?> toMessage(Object mqttMessage, MessageHeaders headers) {
        if (!(mqttMessage instanceof MqttMessage)) {
            throw new IllegalArgumentException("This converter can only convert an 'MqttMessage'; received: " + mqttMessage.getClass().getName());
        } else {
            return this.toMessage((String)null, (MqttMessage)((MqttMessage)mqttMessage));
        }
    }

    public Message<?> toMessage(String topic, MqttMessage mqttMessage) {
        try {
            AbstractIntegrationMessageBuilder<Object> messageBuilder = this.getMessageBuilderFactory().withPayload(this.mqttBytesToPayload(mqttMessage)).setHeader("mqtt_qos", mqttMessage.getQos()).setHeader("mqtt_duplicate", mqttMessage.isDuplicate()).setHeader("mqtt_retained", mqttMessage.isRetained());
            if (topic != null) {
                messageBuilder.setHeader("mqtt_topic", topic);
            }

            return messageBuilder.build();
        } catch (Exception var4) {
            throw new MessageConversionException("failed to convert object to Message", var4);
        }
    }

    public MqttMessage fromMessage(Message<?> message, Class<?> targetClass) {
        byte[] payloadBytes = this.messageToMqttBytes(message);
        MqttMessage mqttMessage = new MqttMessage(payloadBytes);
        Object header = message.getHeaders().get("mqtt_retained");
        Assert.isTrue(header == null || header instanceof Boolean, "mqtt_retained header must be Boolean");
        mqttMessage.setRetained(header == null ? this.defaultRetained : (Boolean)header);
        header = message.getHeaders().get("mqtt_qos");
        Assert.isTrue(header == null || header instanceof Integer, "mqtt_qos header must be Integer");
        mqttMessage.setQos(header == null ? this.defaultQos : (Integer)header);
        return mqttMessage;
    }

    protected Object mqttBytesToPayload(MqttMessage mqttMessage) throws Exception {
        List<Object> list =new ArrayList<Object>();
        list.add(mqttMessage.getPayload());
        return this.payloadAsBytes ? mqttMessage.getPayload() : list;
    }

    protected byte[] messageToMqttBytes(Message<?> message) {
        Object payload = message.getPayload();
        if (!(payload instanceof byte[]) && !(payload instanceof String)) {
            throw new IllegalArgumentException("This default converter can only handle 'byte[]' or 'String' payloads; consider adding a transformer to your flow definition, or subclass this converter for " + payload.getClass().getName() + " payloads");
        } else {
            byte[] payloadBytes;
            if (payload instanceof String) {
                try {
                    payloadBytes = ((String)payload).getBytes(this.charset);
                } catch (Exception var5) {
                    throw new MessageConversionException("failed to convert Message to object", var5);
                }
            } else {
                payloadBytes = (byte[])((byte[])payload);
            }

            return payloadBytes;
        }
    }
}
